package com.fanxl.flink.datastream.batch;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2022/6/1
 */
public class BatchStreamMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings envSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        env.setParallelism(3);
        // note: 增量同步需要开启CK
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, envSettings);

        tableEnvironment.executeSql(" CREATE TABLE demoOrders (\n" +
                "         `order_id` INTEGER ,\n" +
                "          `order_date` DATE ,\n" +
                "          `order_time` TIMESTAMP(3),\n" +
                "          `quantity` INT ,\n" +
                "          `product_id` INT ,\n" +
                "          `purchaser` STRING,\n" +
                "           primary key(order_id)  NOT ENFORCED" +
                "         ) WITH (\n" +
                "          'connector' = 'mysql-cdc',\n" +
                "          'hostname' = '192.168.215.151',\n" +
                "          'port' = '33060',\n" +
                "          'username' = 'totems',\n" +
                "          'password' = '123.Totems#@!',\n" +
                "          'database-name' = 'topology',\n" +
                "          'table-name' = 'demo_orders'," +
                //  全量 + 增量同步
                "          'scan.startup.mode' = 'initial'      " +
                " )");

//        tableEnvironment.executeSql("CREATE TABLE sink (\n" +
//                "         `order_id` INTEGER ,\n" +
//                "          `order_date` DATE ,\n" +
//                "          `order_time` TIMESTAMP(3),\n" +
//                "          `quantity` INT ,\n" +
//                "          `product_id` INT ,\n" +
//                "          `purchaser` STRING,\n" +
//                "          primary key (order_id)  NOT ENFORCED " +
//                ") WITH (\n" +
//                "    'connector' = 'kafka',\n" +
//                "    'properties.bootstrap.servers' = 'localhost:9092',\n" +
//                "    'topic' = 'mqTest02',\n" +
//                "    'format' = 'changelog-json' " +
//                ")");

        tableEnvironment.executeSql("select * from demoOrders").print();
        env.execute();
    }
}


